package com.bdi.mqtt_broker.server;

import com.bdi.mqtt_broker.config.BrokerConfig;
import com.bdi.mqtt_broker.handler.BrokerHandler;
import com.bdi.mqtt_broker.handler.MessageHandler;
import com.bdi.mqtt_broker.handler.ssl.SslContextFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.ssl.SslContext;
import io.netty.util.concurrent.Future;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author WangNing yogehaoren@gmial.com <br>
 * @since 1.0
 */
@Slf4j
@Component
public class BrokerServer {


    /**
     * 存储本Server的全部Channel
     */
    @Autowired
    ChannelGroup channelGroup;

    /**
     * Server 配置文件
     */
    @Autowired
    BrokerConfig brokerConfig;

    /**
     * 协议处理器
     */
    @Autowired
    BrokerHandler brokerHandler;

    /**
     * MQTT Broker Channel
     */
    private Channel serverChannel;

    private final NioEventLoopGroup bossGroup = new NioEventLoopGroup();

    private final NioEventLoopGroup workerGroup = new NioEventLoopGroup();

    private SslContext sslContext = null;



    /**
     * 开启服务器
     * @return serverChannel
     * @throws Exception 启动服务发生异常情况
     */
    public Channel start() throws Exception {



        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workerGroup)
                .channel(isActiveEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {



                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {

                        if(sslContext!=null){

                            ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));

                        }

                        ch.pipeline().addLast("decoder", new MqttDecoder());
                        ch.pipeline().addLast("encoder", MqttEncoder.INSTANCE);
                        ch.pipeline().addLast("broker", brokerHandler);
                        // 以下为业务处理
                        ch.pipeline().addLast("sample", new MessageHandler());

                    }
                })
                .option(ChannelOption.SO_BACKLOG, brokerConfig.getSoBacklog()) // TCP 第三次握手，队列值
                .childOption(ChannelOption.SO_KEEPALIVE, brokerConfig.getKeepAlive());

        ChannelFuture bindFuture = serverBootstrap.bind("0.0.0.0", brokerConfig.getServerPort()).sync();
        log.info("Start MQTT Broker[{}] on {}", brokerConfig.getBrokerId(), brokerConfig.getServerPort());
        serverChannel = bindFuture.channel();
        return bindFuture.channel();

    }


    private static boolean isActiveEpoll() {
        return "linux".equals(System.getProperties().getProperty("os.name"));
    }

    @PostConstruct
    public void initSslContext() throws Exception{

        if(brokerConfig.getSslEnable()){
            log.info("Init SslContext with CrtPath: {}, KeyPath: {}", brokerConfig.getSslCrtPath(), brokerConfig.getSslKeyPath());
            sslContext = SslContextFactory.getSSLContext(brokerConfig.getSslCrtPath(), brokerConfig.getSslKeyPath(), brokerConfig.getSslProvider());
        }
    }

    /**
     * 关闭服务器
     * @throws Exception 关闭服务器发生异常情况
     */
    public void stop() throws Exception{
        if(serverChannel!=null){

            try{
                serverChannel.close().sync();
            }finally {
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }

        }


    }


}
